package com.atguigu.edu.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.MyBoradCastFunction;
import com.atguigu.edu.app.func.MyPhoenixSink;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.util.KafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DimSinkApp {
    public static void main(String[] args) throws Exception {
        //TODO 1、加载环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //TODO 2、状态后端设置
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(6000L);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1500L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS),Time.of(3L,TimeUnit.MINUTES)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 3、读取Kafka中业务主流数据
        String topic="topic_db";
        String groupId="dim_sink_app";
        DataStreamSource<String> eduDS = env.addSource(KafkaUtil.getKafkaConsumer(topic, groupId));

        //TODO 4、主流数据结构的转换
        SingleOutputStreamOperator<JSONObject> JsonDS = eduDS.map(data -> JSON.parseObject(data));

        //TODO 5、主流ETL
        SingleOutputStreamOperator<JSONObject> filterDS = JsonDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                try {
                    jsonObject.getString("data");
                    if (jsonObject.getString("type").equals("bootstrap-start") || jsonObject.getString("type").equals("bootstrap-complete")) {
                        return false;
                    }
                    return true;
                } catch (JSONException jsonException) {
                    return false;
                }
            }
        });

        //TODO 6、读取配置流并广播流
        //6.1 FLinkCDC读取配置表信息
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        //6.2封装为流
        DataStreamSource<String> mysqlDSSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlSource");

        //6.3转成广播配置流
        MapStateDescriptor<String, TableProcess> tableProcessState = new MapStateDescriptor<>("TableProcessState", String.class, TableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlDSSource.broadcast(tableProcessState);

        //TODO 7、链接流
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 8、处理连接流
        SingleOutputStreamOperator<JSONObject> dimDS = connectDS.process(new MyBoradCastFunction(tableProcessState));

        //TODO 9、将数据写入到Phoenix中
        dimDS.addSink(new MyPhoenixSink());

        env.execute();
    }
}
